/*
 * taskpool.h
 *
 *  Created on: Sep 23, 2015
 *      Author: rpi
 */

#ifndef TASKPOOL_H_
#define TASKPOOL_H_

#include <list>
#include <set>
#include <queue>
#include <memory>
#include <iostream>

class task_pool
{
public:
	task_pool(unsigned int pool_size = 128):max_size_(pool_size),stop_all_(true)
	{
	}

	virtual ~task_pool()
	{
	}

	void job_completed( const std::shared_ptr<task> & t)//回收task对象
	{
		std::lock_guard<std::mutex> lock(mutex_);

		bool need_to_notify = idle_tasks_.empty() && (!wait_for_running_jobs_.empty());

		busying_tasks_.erase(t);
		idle_tasks_.push_back(t);

		//LOG(INFO)<<"after job_completed, current idle tasks size:"<< idle_tasks_.size()
		// <<" busying tasks size:"<<busying_tasks_.size()
		// <<" wait for running jobs size:"<<wait_for_running_jobs_.size();
		if( !busying_tasks_.empty() )
		{
			(*busying_tasks_.begin())->print_job();
		}
		if( need_to_notify )//任务太多了，之前空闲线程使用完了，有任务在等待执行，需要通知
		{
			sync_.notify_one();
		}
	};

	//提交任务
	void submit_job( const std::function<void()> & job, const std::string file, int line)
	{
		if( stop_all_ )
		{
			return;
		}

		std::lock_guard<std::mutex> lock(mutex_);
		bool need_notify = wait_for_running_jobs_.empty();
		wait_for_running_jobs_.push(std::make_tuple(job, file, line));

		if( need_notify )//等待执行的任务为空时，需要通知，其他情况不需要通知.
		{
			sync_.notify_one();
		}
	}
	void execute_job()
	{
		while(true)
		{
			std::unique_lock<std::mutex> lock(mutex_);
			while(!stop_all_ && wait_for_running_jobs_.empty() )
			{
				//等待其他线程提交任务
				sync_.wait(lock);
			}

			if( stop_all_ )
			{
				return;
			}

			while(!stop_all_ && idle_tasks_.empty())
			{
				//有任务要执行，但是没有空闲线程，等待其他任务执行完成。
				sync_.wait(lock);
			}

			if( stop_all_ )
			{
				return;
			}

			//有任务，也有空闲线程了
			auto t = get_task();
			auto job = wait_for_running_jobs_.front();
			wait_for_running_jobs_.pop();
			//分发任务到task 线程.
			t->set_job(std::get<0>(job), std::get<1>(job), std::get<2>(job));
		}
	}
	void stop_all()
	{
		std::lock_guard<std::mutex> lock(mutex_);
		stop_all_ = true;
		for( auto t : idle_tasks_ )
		{
			t->stop();
		}

		idle_tasks_.clear();
		for( auto t : busying_tasks_ )
		{
			t->stop();
		}

		while(!wait_for_running_jobs_.empty())
		{
			wait_for_running_jobs_.pop();
		}

		sync_.notify_one();
	}

	void start()
	{// 初始化启动线程池主线程
		try
		{
			std::thread t( [this]{ execute_job(); } );
			t.detach();

			stop_all_ = false;
			allocate_tasks();

		}
		catch( std::exception & e )
		{
		  //LOG(FATAL) << "start tasks pool ... error"<<e.what();
			std::cout << "start tasks pool ... error" << e.what();
		}
	}
protected:
	std::shared_ptr<task> get_task()
	{
		//获取task对象
		if( ! idle_tasks_.empty() )
		{
			auto t = *idle_tasks_.begin();
			idle_tasks_.pop_front();  //从空闲队列移除
			busying_tasks_.insert(t); //加入忙队列

			return t;
		}

		return std::shared_ptr<task>();
	}

	void allocate_tasks() //初始化线程池
	{
		for( int i = 0 ; i < max_size_; i ++ )
		{
			std::shared_ptr<task> t( new task());
			try{
				t->start();
				idle_tasks_.push_back(t);
			}
			catch( std::exception & e)
			{	//超过进程最大线程数限制时，会跑出异常。。。
				break;
			}
		}
	}
private :
	unsigned int	max_size_;
	std::list<std::shared_ptr<task> > idle_tasks_;   //空闲任务队列
	std::set<std::shared_ptr<task> > busying_tasks_;//正在执行任务的队列
	std::queue<std::tuple< std::function<void()>, std::string, int> > wait_for_running_jobs_; //等待执行的任务
	std::mutex	mutex_;
	std::condition_variable	sync_;
	bool stop_all_;
};

#endif /* TASKPOOL_H_ */
